package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/9/13
 * desc:
 **/
public class ProductsDemo {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id       INT,\n" +
                "    name      STRING,\n" +
                "    description   STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
                "  'number-of-rows' = '5', \n" +
//                "  'rows-per-second' = '1', \n" +
                "  'fields.id.min' = '10', \n" +
                "  'fields.id.max' = '200', \n" +
                "  'fields.name.length' = '5', \n" +
                "  'fields.description.length' = '15'\n" +
                ")");

        //Sink
        tableEnv.executeSql("CREATE TABLE sinkTable (" +
                "    id       INT,\n" +
                "    name      STRING,\n" +
                "    description   STRING\n" +
                "   ) " +
                "   WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'products',\n" +
                "  'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
                "  'properties.allow.auto.create.topics' = 'true',\n" +
                "  'sink.parallelism' = '5',\n" +
                "  'format' = 'json'\n" +
//                "  'format' = 'debezium-json'\n" +
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable select * from sourceTable");

    }
}
